package com.zhang.second.day04;

import com.zhang.second.utils.SensorReading;
import com.zhang.second.utils.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @title: 温度传感器连续1s温度上升检测
 * @author: zhang
 * @date: 2022/1/30 20:35
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.sensorId)
                .process(new KeyedProcessFunction<String, SensorReading, String>() {
                    //初始化变量
                    private ValueState<Integer> lastTemp;
                    private ValueState<Long> timerTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //初始化了一张HashMap
                        //key -> Integer
                        lastTemp = getRuntimeContext().getState(
                                new ValueStateDescriptor<Integer>("last-temp", Types.INT)
                        );
                        //初始化了一张HashMap
                        //key -> Long
                        timerTs = getRuntimeContext().getState(
                                new ValueStateDescriptor<Long>("timer-ts", Types.LONG)
                        );
                    }

                    @Override
                    public void processElement(SensorReading value, KeyedProcessFunction<String, SensorReading, String>.Context ctx, Collector<String> out) throws Exception {
                        Integer preTemp = null;
                        if (lastTemp.value() != null) {
                            preTemp = lastTemp.value();
                        }

                        lastTemp.update(value.temperature);

                        //如果是第一条数据
                        if (preTemp == null) {
                            long oneSecondLater = ctx.timerService().currentProcessingTime() + 1000L;
                            ctx.timerService().registerProcessingTimeTimer(oneSecondLater);
                            timerTs.update(oneSecondLater);
                        }

                        Long ts = null;
                        if (timerTs.value() != null) {
                            ts = timerTs.value();
                        }

                        //不是第一条数据
                        if (preTemp != null) {
                            if (value.temperature < preTemp && ts != null) {
                                //如果温度出现下降且定时器存在
                                //1.删除定时器
                                ctx.timerService().deleteProcessingTimeTimer(ts);
                                //2.将保存报警定时器时间戳的状态变量清空
                                timerTs.clear();
                            } else if (value.temperature > preTemp && ts == null) {
                                long oneSecondLater = ctx.timerService().currentProcessingTime() + 1000L;
                                //如果温度上升且没有注册定时器
                                //1 注册1s后的定时器
                                ctx.timerService().registerProcessingTimeTimer(oneSecondLater);
                                //2 保存报警定时器时间戳
                                timerTs.update(oneSecondLater);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("温度传感器：" + ctx.getCurrentKey() + "连续1s温度上升！！！");
                        //将保存报警定时器时间戳的状态变量清空
                        timerTs.clear();
                    }
                })
                .print();

        env.execute();
    }
}
